#pragma once

#ifndef __MY_CONSUMER__
#define __MY_CONSUMER__

#include "../mqcommon/message.pb.h"
#include "../mqcommon/helpper.hpp"

#include <iostream>
#include <string>
#include <memory>
#include <mutex>
#include <functional>
#include <unordered_map>

namespace NS_Consumer
{
    using ConsumerCallback = std::function<void(const std::string &, const NS_Message::MessageProperties *, const std::string &)>;
    struct Consumer
    {
        using Ptr = std::shared_ptr<Consumer>;
        Consumer() {}
        Consumer(const std::string &tag, const std::string &qname, bool auto_ack, ConsumerCallback cb)
            : _tag(tag), _qname(qname), _auto_ack(auto_ack), _callback(std::move(cb))
        {
        }

        ~Consumer() {}
        std::string _tag;
        std::string _qname;
        bool _auto_ack;
        ConsumerCallback _callback;
    };

    class QueueConsumer
    {
    public:
        using Ptr = std::shared_ptr<QueueConsumer>;
        QueueConsumer(const std::string &qname) : _qname(qname), _rr_req(0) {}
        Consumer::Ptr insert(const std::string &tag, const std::string &qname, bool auto_ack, ConsumerCallback cb)
        {
            DLOG("队列单元中插入消费者: %s", tag.c_str());
            std::unique_lock<std::mutex> ul(_mtx);
            DLOG("队列单元中插入消费者检查前: %s", qname.c_str());
            for (auto &consumer : _consumers)
            {
                if (consumer->_tag == tag)
                {   
                    DLOG("消费者创建重复: %s - %s", qname.c_str(), tag.c_str());
                    return consumer;
                }
            }

            DLOG("队列单元中插入消费者开始: %s", qname.c_str());
            Consumer::Ptr ptr = std::make_shared<Consumer>(tag, qname, auto_ack, cb);
            DLOG("队列单元中插入消费者完成: %s", qname.c_str());
            _consumers.push_back(ptr);
            return ptr;
        }
        void erase(const std::string &tag)
        {
            std::unique_lock<std::mutex> ul(_mtx);
            for (auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if ((*it)->_tag == tag)
                {
                    _consumers.erase(it);
                    return;
                }
            }
        }
        Consumer::Ptr choose()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            if (_consumers.size() == 0)
            {
                return Consumer::Ptr();
            }

            int idx = _rr_req++ % _consumers.size();
            return _consumers[idx];
        }

        bool exists(const std::string &tag)
        {
            std::unique_lock<std::mutex> ul(_mtx);
            for (auto &consumer : _consumers)
            {
                if (consumer->_tag == tag)
                    return true;
            }
            return false;
        }
        bool empty()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            return _consumers.size() == 0;
        }
        void clear()
        {
            std::unique_lock<std::mutex> ul(_mtx);
            _consumers.clear();
            _rr_req = 0;
        }

        ~QueueConsumer() {}

    private:
        std::mutex _mtx;
        std::string _qname;

        std::vector<Consumer::Ptr> _consumers;
        uint64_t _rr_req;
    };

    class ConsumerManager
    {
    public:
        using Ptr = std::shared_ptr<ConsumerManager>;
        ConsumerManager() {}
        bool initQueueConsumer(const std::string qname)
        {
            std::unique_lock<std::mutex> ul(_mtx);
            if (_queues.find(qname) != _queues.end())
            {
                DLOG("消费者队列单元创建重复: %s", qname.c_str());
                return true;
            }

            _queues[qname] = std::make_shared<QueueConsumer>(qname);
            return true;
        }
        void destoryQueueConsumer(const std::string qname)
        {
            QueueConsumer::Ptr ptr;
            {
                std::unique_lock<std::mutex> ul(_mtx);
                if (_queues.find(qname) == _queues.end())
                    return;
                ptr = _queues[qname];
                _queues.erase(qname);
            }

            ptr->clear();
        }

        Consumer::Ptr insert(const std::string &qname, const std::string &tag, bool auto_ack, const ConsumerCallback &cb)
        {
            {
                std::unique_lock<std::mutex> ul(_mtx);
                if (_queues.find(qname) != _queues.end() && _queues[qname]->exists(tag))
                {
                    DLOG("消费者创建重复: %s - %s", qname.c_str(), tag.c_str());
                    return Consumer::Ptr();
                }
            }
            DLOG("消费者创建: %s - %s", qname.c_str(), tag.c_str());
            return _queues[qname]->insert(tag, qname, auto_ack, cb);
        }
        void erase(const std::string &qname, const std::string &tag)
        {
            {
                std::unique_lock<std::mutex> ul(_mtx);
                if (_queues.find(qname) == _queues.end())
                    return;
            }

            _queues[qname]->erase(tag);
        }

        void eraseAll(const std::string &qname)
        {
            {
                std::unique_lock<std::mutex> ul(_mtx);
                if (_queues.find(qname) == _queues.end())
                    return;
            }

            _queues.erase(qname);
        }

        Consumer::Ptr choose(const std::string &qname)
        {
            {
                std::unique_lock<std::mutex> ul(_mtx);
                if (_queues.find(qname) == _queues.end())
                {
                    ELOG("没有对应的消费者队列：%s", qname.c_str());
                    return Consumer::Ptr();
                }
            }

            return _queues[qname]->choose();
        }

        bool exists(const std::string &qname, const std::string &tag)
        {
            {
                std::unique_lock<std::mutex> ul(_mtx);
                if (_queues.find(qname) == _queues.end())
                {
                    ELOG("没有对应的消费者队列：%s", qname.c_str());
                    return false;
                }
            }
            return _queues[qname]->exists(tag);
        }
        bool empty(const std::string &qname)
        {
            {
                std::unique_lock<std::mutex> ul(_mtx);
                if (_queues.find(qname) == _queues.end())
                {
                    ELOG("没有对应的消费者队列：%s", qname.c_str());
                    return false;
                }
            }
            return _queues[qname]->empty();
        }
        void clear(const std::string &qname)
        {
            _queues.clear();
        }

        ~ConsumerManager() {}

    private:
        std::mutex _mtx;
        std::unordered_map<std::string, QueueConsumer::Ptr> _queues;
    };
}

#endif